package server;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;

import server.Message.MsgType;

/**
 * This thread takes care of handling data from the users/clients.
 */
public class UserRecvThread extends Thread {
	/* Mapping from pubID -> sockaddr */
	private ConcurrentHashMap<String, SocketAddress> uAddrs;
	/* Address to liveness */
	private SocketAddress liveAddr;

	/* Queues to put messages on */
	private BlockingQueue<Message> uOutgoing;  // client
	private BlockingQueue<Message> sOutgoing;  // server
	private BlockingQueue<Message> connections; // connections 

	private Map<String, CopyOnWriteArrayList<SocketAddress>> sAddrs;
	private ConcurrentHashMap<String, Long> lastSeq;
	
	/* Network junk */
	private Selector select;
	private DatagramChannel sock;
	public boolean alive = true;

	private ByteBuffer in = ByteBuffer.allocate(2048);

	public UserRecvThread(BlockingQueue<Message> uOutgoing,
			ConcurrentHashMap<String, SocketAddress> uAddrs,
			DatagramChannel uSock, BlockingQueue<Message> sOutgoing,
			Map<String, CopyOnWriteArrayList<SocketAddress>> sAddrs,
			int livePort, BlockingQueue<Message> connections,
			ConcurrentHashMap<String, Long> lastSeq) {
		super("UserRecv-" + uSock.socket().getLocalPort());
		this.uOutgoing = uOutgoing;
		this.sOutgoing = sOutgoing;
		this.connections = connections;
		this.lastSeq = lastSeq;
		
		this.uAddrs = uAddrs;
		this.sAddrs = sAddrs;
		this.sock = uSock;
		this.liveAddr = new InetSocketAddress("localhost", livePort);
	}

	private void log(String msg) {
		Server.log(this.getName(), msg);
	}

	@Override
	public void run() {
		try {
			select = Selector.open();
			sock.register(select, SelectionKey.OP_READ);

			while (alive) {
				try {
					select.select();
				} 
				catch (IOException e) {
					e.printStackTrace();
					// TODO select error'd
				}

				for (SelectionKey s : select.selectedKeys()) {
					if (s.isValid() && s.isReadable()) {
						receive(s);
						select.selectedKeys().remove(s);
					}
				}
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	/**
	 * Handles the data to be published from the user.
	 * 
	 * @param s
	 *            The 'key' (the listening socket) to be read.
	 */
	private void receive(SelectionKey s) {
		try {
			in.clear();
			DatagramChannel ch = ((DatagramChannel) s.channel());
			SocketAddress addr = ch.receive(in);
			in.flip();

			Message msg = Message.deserialize(in);

			/* With a PUB. */
			if (msg.type == MsgType.PUB) {
				uAddrs.put(msg.pubId, addr);
				lastSeq.put(msg.pubId, msg.seq);
				sOutgoing.put(msg);
			}
			
			/* With a SUB, we need to 'join' the mesh by forcing a join on liveness */
			else if (msg.type == MsgType.SUB) {
				if (!sAddrs.containsKey(msg.data)) {
					sAddrs.put(msg.data, Server.initMesh());
					notifyLiveness(msg.data);
				}
				uAddrs.put(msg.pubId, addr);
			}
			else if (msg.type == MsgType.CONNECT) {
				msg.addr = addr;
				
				if ((msg.flags & 0x1) == 1) {  // new publisher
					sAddrs.put(msg.pubId, Server.initMesh());
					connections.put(msg);
				}
				else {  //new subscriber
					connections.put(msg);
				}
				return;
			}
			// any and all messages should go to the user send thread
			uOutgoing.put(msg);

		} catch (IOException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
	
	private void notifyLiveness(String id) {
		log("Notifying liveness to join '" + id + "'");
		try {
			Message notify = new Message(MsgType.REQUEST, id, 0, "");
			sock.send(notify.serialize(), liveAddr);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
}
